package dht;
import dht.message.Message;
import dht.message.MessageUtils;
import dht.message.PaxosMessage;
import dht.paxos.Paxos;
import java.io.File;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class Server {
    
    public static void main(String[] args) throws Exception {
        if (args.length < 1) {
            System.out.println("Usage: java Server server-name");
            System.exit(1);
        }
        
        HostAddress server = new HostAddress(args[0], Configuration.UDP_PORT);
        new Server(server).start();
        // maybe some logic to let the user type 'exit' to shut down the server
        // later on
    }
    
    
    private static final int TIMEOUT = 2000; // ms
    private static final int MAX_TIMEOUTS = 5;
    private static final int MAX_PAXOS_RETRIES = 5;
    private final HostAddress localAddress;
    private final int localport;
    private final ScheduledExecutorService pool;
    private final DatagramSocket socket;
    private DatagramSocketListener socklistener;
    private TcpListener tcpListener;
    private Paxos paxos;

    public Server(HostAddress localAddress) throws SocketException, IOException {
        this.localAddress = localAddress;
        this.localport = localAddress.port();
        this.pool = Executors.newScheduledThreadPool(5);
        this.socket = new DatagramSocket(localport);
        this.socket.setSoTimeout(TIMEOUT);
    }

    public HostAddress getLocalAddress() { 
        return localAddress;
    }
    
    /**
     * Starts the services running on this node
     */
    public void start() throws SocketException, IOException {
        if (socklistener == null || !socklistener.isRunning) {
            socklistener = new DatagramSocketListener(this);
            pool.execute(socklistener);
        }
        
        if (tcpListener == null || !tcpListener.running) {
            tcpListener = new TcpListener(this);
            pool.execute(tcpListener);
        }
        paxos = new Paxos(this);
    }

    /**
     * Stops the services running on this node
     */
    public void stop() {
        socklistener.stop();
        pool.shutdown();
        // socket.close();
    }

    public void updateDatabase(Star s) {
        PutResponse response = Database.getInstance().put(new PutRequest(s));
    }

    /**
     * Sends the Message to Remote Node <p>
     *
     * @param sendData - Data to send to HostAddress
     */
    public void sendMessage(Message aMessage) {
        try {
            byte[] message = aMessage.getBytes();
            DatagramPacket p = createDatagram(message, aMessage.getAddress());
            socket.send(p);
        } catch (IOException ex) {
            // until we decide how to handle
            throw new RuntimeException(ex);
        }
    }

    public void sendReliably(Message m) {
        sendMessage(m);
        scheduleResender(new MessageResender(this, m));
    }

    private void scheduleResender(MessageResender resender) {
        pool.schedule(resender, TIMEOUT, TimeUnit.MILLISECONDS);
    }

    private DatagramPacket createDatagram(byte[] b, HostAddress addr) {
        return new DatagramPacket(b, b.length, addr.address(), addr.port());
    }

    public Message receiveMessage() throws IOException {
        byte[] receiveData = new byte[Message.MAX_LENGTH];
        DatagramPacket packet = new DatagramPacket(receiveData, receiveData.length);
        socket.receive(packet);
        return MessageUtils.createMessage(packet.getData(), new HostAddress(packet.getAddress(), packet.getPort()));
    }

    protected void messageReceived(Message aMessage) {
        System.out.println("Received message " + aMessage.getOpCode());
        if (aMessage instanceof PaxosMessage)
            paxos.messageReceived((PaxosMessage)aMessage);
    }
    
    private PutResponse handlePutRequest(PutRequest request) {
        boolean succeeded = false;
        for (int i=0; i<MAX_PAXOS_RETRIES; i++) {
            try {
                succeeded = paxos.atomicBroadcast(request.star);
                if (succeeded)
                    break;
            } catch (InterruptedException ex) {
                continue;
            }
        }
        if (!succeeded)
            return new PutResponse(succeeded? Status.Worked : Status.WriteError);
        return Database.getInstance().put(request);
    }

    private GetResponse handleGetRequest(GetRequest request) {
        return Database.getInstance().get(request);
    }
    
    private boolean shouldResend(Message m) {
        return paxos.shouldResend(m);
    }

    private void hostDied(HostAddress host) {
        paxos.hostDied(host);
    }

    /**
     * Inner Handler class for service, to process Datagram packets and react
     * according to type. Wraps byte array into a ByteBuffer for easier
     * handling.
     */
    public class DatagramSocketListener
            implements Runnable {

        private volatile boolean isRunning = true;
        private final Server connection;

        DatagramSocketListener(Server c) {
            this.connection = c;
        }

        @Override
        public void run() {
            System.out.println("Starting DHT service on port: " + localport);
            while (isRunning) {
                final Message message;
                try {
                    message = connection.receiveMessage();
                } catch (IOException ex) {
                    if (ex instanceof SocketTimeoutException) {
                        continue;
                    } else {
                        throw new RuntimeException(ex);
                    }
                }

                connection.messageReceived(message);
            }
        }

        public void stop() {
            isRunning = false;
        }
    }

    private final class MessageResender implements Runnable {

        private final Server n;
        private final Message m;
        private int timeouts = 0;

        MessageResender(Server n, Message m) {
            this.n = n;
            this.m = m;
        }

        @Override
        public void run() {
            if (!n.shouldResend(m) || timeouts >= MAX_TIMEOUTS) {
                if (timeouts >= MAX_TIMEOUTS) {
                    n.hostDied(m.getAddress());
                }
                return;
            }

            ++timeouts;
            System.out.println("resending " + m.getOpCode() + " timeouts = " + timeouts);
            n.sendMessage(m);
            n.scheduleResender(this);
        }
    }
    
    private final class TcpListener implements Runnable {
        private volatile boolean running = true;
        public void stop() { running = false; }
        public void start() { running = true; }
        
        private final Server connection;
        private final ServerSocket socket;
        
        public TcpListener(Server c) throws IOException {
            this.connection = c;
            this.socket = new ServerSocket(Configuration.TCP_SERVER_PORT);
        }

        @Override
        public void run() {
            while (running) {
                try (Socket client = socket.accept()) {
                    try (ObjectOutputStream oos = new ObjectOutputStream(client.getOutputStream())) {
                        try (ObjectInputStream ois = new ObjectInputStream(client.getInputStream())) {
                            Object query;
                            try {
                                query = ois.readObject();
                            } catch (ClassNotFoundException ex) {
                                System.out.println("Received invalid tcp request");
                                continue;
                            }

                            Query response = null;
                            if (query instanceof PutRequest) {
                                System.out.println("received put request ...");
                                response = handlePutRequest((PutRequest)query);
                                System.out.println("writing response ... ");
                                oos.writeObject(response);
                            }
                            else if (query instanceof GetRequest) {
                                System.out.println("received get request");
                                response = handleGetRequest((GetRequest)query);
                                System.out.println("writing response ... ");
                                oos.writeObject(response);
                            }
                            else {
                                System.out.println("Received invalid tcp request");
                                continue;
                            }
                            
                            oos.flush();
                            System.out.println("done with message ... ");
                        }
                    }
                    
                } catch (Exception ex) {
                    ex.printStackTrace();
                    throw new RuntimeException(ex);
                }
            }
        }
    }
}
